#include <iostream>
#include <thread>
#include <sstream>
#include <fstream>
#include "sockpp/tcp_acceptor.h"
#include "stream.pb.h"
#include "OpusDecoder.h"
#include <netinet/tcp.h>

#define PCM_RECORD

//#define LOG_DURATION(n)
#define LOG_DURATION(n) (std::cout << n << std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now() - start).count() << std::endl)

void static run_opus_stream(sockpp::tcp_socket sock) {
    ssize_t n;
    //TODO: init asr session resources
    yd::audioprocess::OpusDecoder decoder;
    int yes = 1;
    sock.set_option(IPPROTO_TCP, TCP_NODELAY, (char *) &yes, sizeof(int));
#ifdef PCM_RECORD
    std::ofstream rec(sock.peer_address().to_string().data(), std::ios_base::out | std::ios_base::binary);
    if (!rec.is_open()) {
        std::cerr << "open pcm record file failed" << strerror(errno) << std::endl;
    }
#endif
    int count = 0;
    while (true) {
        //read size
        int32_t len = 0;
        n = sock.read_n((void *) &len, 4);
        if (n != 4)
            break;

        //read protobuf package
        std::string s;
        s.resize(len);
        n = sock.read_n(&s[0], len);
        if (n != len)
            break;

        //decode to protobuf struct
        yd::audioprocess::Opus pkg;
        auto start = std::chrono::system_clock::now();
        if (!pkg.ParseFromString(s)) {
            std::cerr << "parse from package failed" << std::endl;
        }
        LOG_DURATION("decode protobuf");


        //decode opus to pcm
        std::string pcm;
        if (pkg.pkg().size() == 0) {
            pcm.resize(0);
        } else {
            auto start = std::chrono::system_clock::now();
            decoder.Decode(pkg.pkg(), &pcm);
            LOG_DURATION("decode opus use ");
#ifdef PCM_RECORD
            rec.write(pcm.data(), pcm.size());
#endif
        }

        //TODO: process pcm.data() & pcm.size() to asr session. here is example sending back received size
        {
            if (pcm.size() > 0) {
                yd::audioprocess::AsrResult result;
                std::stringstream ss;
                std::string outStr;
                ss << pcm.size() << '[' << count++ << ']' << std::chrono::time_point_cast<std::chrono::milliseconds>(
                        std::chrono::system_clock::now()).time_since_epoch().count();
                result.set_text(ss.str());
                auto start = std::chrono::system_clock::now();
                result.SerializeToString(&outStr);
                LOG_DURATION("encode protobuf use ");
                int32_t len = outStr.size();
                auto wr = sock.write_n(&len, 4);
                if (wr != 4) {
                    break;
                }
                wr = sock.write_n(&outStr[0], outStr.size());
                if (wr != outStr.size())
                    break;
            } else {
                len = 0;
                sock.write_n(&len, 4);
            }
        }
    }
    std::cout << "Connection closed from " << sock.peer_address() << std::endl;
    //TODO: clean up asr session resources
}

int main() {
    sockpp::socket_initializer sockInit;
    const auto port = 3990;
    sockpp::tcp_acceptor acc(port);
    if (!acc) {
        std::cerr << "Error creating the acceptor: " << acc.last_error_str() << std::endl;
        return 1;
    }
    std::cout << "Awaiting connections on port " << port << "..." << std::endl;

    while (true) {
        sockpp::inet_address peer;

        // Accept a new client connection
        sockpp::tcp_socket sock = acc.accept(&peer);
        std::cout << "Received a connection request from " << peer << std::endl;

        if (!sock) {
            std::cerr << "Error accepting incoming connection: "
                      << acc.last_error_str() << std::endl;
        } else {
            // Create a thread and transfer the new stream to it.
            std::thread thr(run_opus_stream, std::move(sock));
            thr.detach();
        }
    }
    return 0;
}
